package com.cetcs.kmga.dataManager.util;

import com.google.common.collect.Lists;
import com.rabbitmq.client.*;
import org.apache.commons.lang.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.TimeoutException;

/**
 * Mq消费者工具类
 *
 * @author xutao
 * @version V1.0 创建时间：2017-10-13 15:59
 * Copyright 2017 by CETC
 */
public class ConsumerMqUtil extends Thread {

    private final static Logger LOGGER = LoggerFactory.getLogger(ConsumerMqUtil.class);

    /**
     * 域名
     */
    private String MQ_HOST;

    /**
     * 用户名
     */
    private String MQ_USER;

    /**
     * 密码
     */
    private String MQ_PWD;

    /**
     * mq的端口
     */
    private String MQ_PORT;

    private String MQ_NODE_LIST;

    private final static String MQ_URL = "conf/MqService.properties";

    private ConnectionFactory factory;

    private Connection connection;

    private Channel channel;

    /**
     * mq消息操作类
     */
    private IConsumerMqMsgOper mqMsgOper;

    /**
     * 队列名称
     */
    private String queueName;


    {
        MQ_HOST = ReadConfigUtil.getValueByKey(MQ_URL, "MQ_HOST");
        MQ_USER = ReadConfigUtil.getValueByKey(MQ_URL, "MQ_USER");
        MQ_PWD = ReadConfigUtil.getValueByKey(MQ_URL, "MQ_PWD");
        MQ_PORT = ReadConfigUtil.getValueByKey(MQ_URL, "MQ_PORT");
        MQ_NODE_LIST = ReadConfigUtil.getValueByKey(MQ_URL, "MQ_NODE_LIST");
        factory = new ConnectionFactory();
//        factory.setHost(MQ_HOST);
        factory.setUsername(MQ_USER);
        factory.setPassword(MQ_PWD);
//        if (StringUtils.isNotBlank(MQ_PORT)) {
//            factory.setPort(Integer.parseInt(MQ_PORT));
//        }
        try {
            if (StringUtils.isNotBlank(MQ_NODE_LIST)) {
                List<Address> addressList = Lists.newArrayList();
                String[] nodeArr = MQ_NODE_LIST.split(",");
                for (int i = 0; i < nodeArr.length; i++) {
                    String[] nodeStr=nodeArr[i].split(":");
                    Address address=null;
                    if(nodeStr.length>1){
                        address=new Address(nodeStr[0],Integer.parseInt(nodeStr[1]));
                    }else {
                        address=new Address(nodeStr[0]);
                    }
                    addressList.add(address);
                }
                connection = factory.newConnection(addressList);
            } else {
                connection = factory.newConnection();
            }
        } catch (IOException e) {
            LOGGER.error("创建MQ连接失败，失败原因：" + e.getMessage());
        } catch (TimeoutException e) {
            LOGGER.error("创建MQ连接超时，超时原因：" + e.getMessage());
        }
    }


    @Override
    public void run() {
        DefaultConsumer consumer;
        try {
            channel = connection.createChannel();
        } catch (IOException e) {
            e.printStackTrace();
        }
        consumer = new DefaultConsumer(channel) {
            @Override
            public void handleDelivery(String consumerTag,
                                       Envelope envelope,
                                       AMQP.BasicProperties properties,
                                       byte[] body) throws IOException {
                try {
                    String message = new String(body, "UTF-8");
                    mqMsgOper.operBackMsg(message);
                    channel.basicAck(envelope.getDeliveryTag(), false);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    rejectMessage(envelope.getDeliveryTag());
                }
            }
        };
        try {
            channel.basicConsume(queueName, false, consumer);
        } catch (IOException e) {
            e.printStackTrace();
            this.close();
        }
    }

    private void rejectMessage(long deliveryTag) {
        try {
            channel.basicReject(deliveryTag, false);
        } catch (IOException ex) {
            ex.printStackTrace();
        }
    }

    private void close() {
        try {
            if (channel != null) {
                channel.close();
                channel = null;
            }
            if (connection != null) {
                connection.close();
                connection = null;
            }
        } catch (IOException | TimeoutException ex) {
            ex.printStackTrace();
        }
    }

    public String getQueueName() {
        return queueName;
    }

    public void setQueueName(String queueName) {
        this.queueName = queueName;
    }

    public IConsumerMqMsgOper getMqMsgOper() {
        return mqMsgOper;
    }

    public void setMqMsgOper(IConsumerMqMsgOper mqMsgOper) {
        this.mqMsgOper = mqMsgOper;
    }
}
